package com.winit.schedule.job.tracker.remoting;

import io.netty.channel.Channel;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.winit.schedule.core.conf.Configuration;
import com.winit.schedule.core.exception.RemotingSendException;
import com.winit.schedule.core.service.AbstractService;
import com.winit.schedule.job.tracker.JobTrackerContext;
import com.winit.schedule.job.tracker.conf.JobConfiguration;
import com.winit.schedule.remoting.InvokeCallback;
import com.winit.schedule.remoting.RemotingServer;
import com.winit.schedule.remoting.exception.RemotingCommandFieldCheckException;
import com.winit.schedule.remoting.netty.NettyRemotingServer;
import com.winit.schedule.remoting.netty.NettyRequestProcessor;
import com.winit.schedule.remoting.netty.NettyServerConfig;
import com.winit.schedule.remoting.protocol.RemotingCommand;

public class RemotingService extends AbstractService {

    private static final Logger log = LoggerFactory.getLogger(RemotingService.class);

    private RemotingServer      remotingServer;

    private long                invokeTimeoutMillis;

    private JobTrackerContext   jobTrackerContext;

    public RemotingService(JobTrackerContext jobTrackerContext){
        super("RemotingService");
        this.jobTrackerContext = jobTrackerContext;
    }

    public RemotingService(String name){
        super(name);
    }

    public void registerProcessor(int requestCode, NettyRequestProcessor processor, ExecutorService executor) {
        remotingServer.registerProcessor(requestCode, processor, executor);
    }

    public void registerDefaultProcessor(NettyRequestProcessor processor, ExecutorService executor) {
        remotingServer.registerDefaultProcessor(processor, executor);
    }

    public RemotingCommand invokeSync(Channel channel, RemotingCommand request) throws RemotingSendException,
                                                                               RemotingCommandFieldCheckException {
        try {

            request.checkCommandBody();

            return remotingServer.invokeSync(channel, request, invokeTimeoutMillis);
        } catch (RemotingCommandFieldCheckException e) {
            throw e;
        } catch (Throwable t) {
            throw new RemotingSendException(t);
        }
    }

    public void invokeAsync(Channel channel, RemotingCommand request, InvokeCallback invokeCallback)
                                                                                                    throws RemotingCommandFieldCheckException,
                                                                                                    RemotingSendException {
        try {

            request.checkCommandBody();

            remotingServer.invokeAsync(channel, request, invokeTimeoutMillis, invokeCallback);
        } catch (RemotingCommandFieldCheckException e) {
            throw e;
        } catch (Throwable t) {
            throw new RemotingSendException(t);
        }
    }

    public void invokeOneway(Channel channel, RemotingCommand request) throws RemotingCommandFieldCheckException,
                                                                      RemotingSendException {
        try {

            request.checkCommandBody();

            remotingServer.invokeOneway(channel, request, invokeTimeoutMillis);
        } catch (RemotingCommandFieldCheckException e) {
            throw e;
        } catch (Throwable t) {
            throw new RemotingSendException(t);
        }
    }

    @Override
    protected void serviceInit(Configuration conf) throws Exception {

        invokeTimeoutMillis = this.getConfig().getLong(JobConfiguration.CONNECT_TIMEOUT,
            JobConfiguration.DEFAULT_CONNECT_TIMEOUT);

        log.info("initail connect timeout is : " + invokeTimeoutMillis + "ms");

        int listenPort = this.getConfig().getInt(JobConfiguration.LISTEN_PORT, JobConfiguration.DEFAULT_LISTEN_PORT);

        NettyServerConfig config = new NettyServerConfig();
        // config 配置
        config.setListenPort(listenPort);

        remotingServer = new NettyRemotingServer(config);

        super.serviceInit(conf);
    }

    /**
     * 注册默认请求处理器，分发请求
     */
    public void registerDefaultProcessor() {
        NettyRequestProcessor defaultProcessor = new RemotingDispatcher(this);
        remotingServer.registerDefaultProcessor(defaultProcessor, Executors.newCachedThreadPool());
    }

    @Override
    protected void serviceStart() throws Exception {
        remotingServer.start();
        registerDefaultProcessor();
        super.serviceStart();
    }

    @Override
    protected void serviceStop() throws Exception {
        remotingServer.shutdown();
        super.serviceStop();
    }

    public JobTrackerContext getJobTrackerContext() {
        return jobTrackerContext;
    }

    public RemotingServer getRemotingServer() {
        return remotingServer;
    }

}
